package streammuxer

import (
	"fmt"
	"sync/atomic"

	"gitee.com/general252/gortc/pkg/util"
)

type Stream struct {
	info    StreamInfo
	pullers *util.Map[pullId, *Puller]

	cache        atomic.Int32 // 0未加载, 1有数据, -1无变化
	cachePullers []*Puller
}

func newStream(info StreamInfo) *Stream {
	return &Stream{
		info:         info,
		pullers:      util.NewMap[pullId, *Puller](),
		cachePullers: []*Puller{},
	}
}

func (st *Stream) close() {
	for _, puller := range st.GetPullers() {
		puller.OnStreamClose(st)
	}
}

func (st *Stream) write(data *StreamData) {
	for _, puller := range st.GetPullers() {
		puller.OnPacket(&st.info, data)
	}
}

func (st *Stream) AddPull(puller *Puller) error {
	if puller.pullId == 0 {
		return fmt.Errorf("error pullId")
	}

	st.pullers.Store(puller.pullId, puller)
	st.cache.Store(1)

	if st.info.OnNewPuller != nil {
		st.info.OnNewPuller(puller)
	}
	return nil
}

func (st *Stream) RemotePull(puller *Puller) {
	st.pullers.Delete(puller.pullId)

	st.cache.Store(1)

	if st.info.OnDelPuller != nil {
		st.info.OnDelPuller(puller)
	}
}

func (st *Stream) GetInfo() *StreamInfo {
	return &st.info
}

func (st *Stream) GetPullers() []*Puller {
	value := st.cache.Load()

	if value == 0 || value == 1 {
		st.cachePullers = st.cachePullers[:0]
		st.pullers.Range(func(key pullId, value *Puller) bool {
			st.cachePullers = append(st.cachePullers, value)
			return true
		})

		st.cache.CompareAndSwap(value, -1)
	}

	return st.cachePullers
}

func (st *Stream) String() string {
	return fmt.Sprintf("stream{id: %v, name: %v}", st.info.Id, st.info.Name)
}
